package org.niit.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object ActionRDD02 {
  //算子：就是RDD自带的一些方法
  //行动算子：将RDD中的元素进行聚合 遍历 返回 统计出现次数 保存文件
  def main(args: Array[String]): Unit = {
    //准备环境
    // -- 创建Spark的配置                             CPU核数等于分区数
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark") //重点
    //创建SparkContext 去加载配置
    val sparkContext = new SparkContext(sparkConf)//重点
    //设置日志级别
    sparkContext.setLogLevel("ERROR")

    // 1.reduce:聚合RDD中的所有元素，先聚合分区内数据，再聚合分区外数据
    val rdd1: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4), 2)// 【1，2】，【3，4】
    val i: Int = rdd1.reduce(_ + _)   // 1+2   3+4     3 + 7 = 10
    println(i)

    //2.collect: 在driver中，以数组的行为返回数据集的所有元素。 将RDD变成数组
    val arrs1: Array[Int] = rdd1.collect()
    println(arrs1.mkString(","))

    //3.count:获得RDD元素的个数
    val count: Long = rdd1.count()
    println("当前RDD元素个数：",count)

    //4.first:返回RDD中的第一个元素
    val first: Int = rdd1.first()
    println(first)

    //5.take:返回一个由RDD前n个元素组成数组
    val arrs2: Array[Int] = rdd1.take(3)
    println(arrs2.mkString(","))

    //6.takeOrdered: 返回该RDD排序后的前n个元素组成的数组
    val rdd2: RDD[Int] = sparkContext.makeRDD(List(3, 1, 2, 4, 7))
    val arrs3: Array[Int] = rdd2.takeOrdered(2)
    println(arrs3.mkString(","))

   //7.aggregate: 分区的数据通过  初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合
    val res1: Int = rdd1.aggregate(10)(_ + _, _ + _) //【1,2,10】=13  【3，4，10】 = 17    13 + 17 + 10
    println(res1)

    val res2: Int = rdd1.fold(10)(_ + _)
    println(res2) //【1,2,10】=13  【3，4，10】 = 17    13 + 17 + 10

    //8.countByKey:统计每种Key出现的次数
    val rdd3: RDD[(String, Int)] = sparkContext.makeRDD(List( ("a", 1), ("a", 2), ("b", 3)))
    val map: collection.Map[String, Long] = rdd3.countByKey()
    println(map)

    /*
    9. save相关算子
          saveAsTextFile 任意类型
          saveAsObjectFile 任意类型  将数据保存为对象格式
          saveAsSequenceFile 数据格式必须是k-v类型
     */
    rdd3.saveAsTextFile("output1")
    rdd3.saveAsObjectFile("output2")
    rdd3.saveAsSequenceFile("output3")

    sparkContext.stop()
  }
}
